package com.glzn.springboot03.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.glzn.springboot03.entity.Mqtt;
import com.glzn.springboot03.util.ByteUtil;
import com.glzn.springboot03.util.ComService;
import com.glzn.springboot03.util.Frame;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @author gongjq
 */
public class MqttPublishClient implements MqttCallback{
	private static final Logger logger = LoggerFactory.getLogger(MqttPublishClient.class);

	private static final int MAX_FRAME_LENGTH = 1200;

	private static final int MAX_FRAME_COUNT = 127;

	private int frameCount = 1;

	private byte[] recvData;

	private byte[] payload;

	private int recvLen;

	private int totalLen;

	private Mqtt mqtt;

	private int tryTimes = 3;

	private MqttClient mqttClient;

	public MqttPublishClient() {
	}

	public Mqtt getMqtt() {
		return mqtt;
	}

	public void setMqtt(Mqtt mqtt) {
		this.mqtt = mqtt;
	}

	public void getInstance(Mqtt mqtt) {
		String timestamp = String.valueOf(System.currentTimeMillis());
		String clientId = "java" + System.currentTimeMillis();
		String hostUrl = "tcp://" + mqtt.getProductKey() + ".iot-as-mqtt." + mqtt.getRegionId() + ".aliyuncs.com:1883";
		String mqttclientId = clientId + "|securemode=3,signmethod=hmacsha1,timestamp=" + timestamp + "|";
		String username = mqtt.getDeviceName() + "&" + mqtt.getProductKey();

		Map<String, String> params = new HashMap<>(16);
		params.put("productKey", mqtt.getProductKey());
		params.put("deviceName", mqtt.getDeviceName());
		params.put("clientId", clientId);
		params.put("timestamp", timestamp);
		String password = MqttConfig.sign(params, mqtt.getDeviceSecret(), "hmacsha1");

		connect(hostUrl, mqttclientId, username, password, mqtt.getTimeout(), mqtt.getKeepalive());

		String subTopic = mqtt.getProductKey() + "/" + mqtt.getDeviceName() + "/user/get";

		subscribe(subTopic, 1);

		this.setMqtt(mqtt);
	}

	public MqttClient getMqttClient() {
		return mqttClient;
	}

	public void setMqttClient(MqttClient mqttClient) {
		this.mqttClient = mqttClient;
	}

	public void connect(String hostUrl, String clientId, String username, String password, int timeout, int keepalive){
		MqttClient client;
		try {
			client = new MqttClient(hostUrl,clientId, new MemoryPersistence());
			MqttConnectOptions options = new MqttConnectOptions();
			options.setCleanSession(true);
			options.setMqttVersion(4);
			options.setUserName(username);
			options.setPassword(password.toCharArray());
			options.setConnectionTimeout(timeout);
			options.setKeepAliveInterval(keepalive);
			setMqttClient(client);
			//设置回调函数
			client.setCallback(this);
			client.connect(options);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	public void publish(String topic, String payload, int qos, boolean retained){
		/*
		MqttMessage mqttMessage = new MqttMessage();
		mqttMessage.setQos(qos);
		mqttMessage.setRetained(retained);
		mqttMessage.setPayload(payload.getBytes());

		MqttTopic mtopic = getMqttClient().getTopic(topic);

		if (null == mtopic){
			logger.error("topic not exist");
		}
		MqttDeliveryToken token;
		try {
			assert mtopic != null;
			token = mtopic.publish(mqttMessage);
			token.waitForCompletion();
		} catch (MqttException e) {
			e.printStackTrace();
		}*/
		try {
			logger.debug("发送数据中:{}", Thread.currentThread().getId());
			getMqttClient().publish(topic, payload.getBytes(), qos, retained);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	public void subscribe(String topic, int qos){
		logger.info("开始订阅主题:{}", topic);
		try {
			getMqttClient().subscribe(topic, qos);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void connectionLost(Throwable throwable) {
		logger.info(mqtt.getId()+"断开连接,准备重连");
		if (null != mqttClient && tryTimes != 0){
			getInstance(mqtt);
			tryTimes--;
		}
	}

	@Override
	public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
	  //1、将接收的数据转为json格式
		JSONObject object = (JSONObject) JSONObject.parse(mqttMessage.getPayload());
		//2、获取message字段的数据
		String message = object.getString("message");
		//4、获取len的长度
		Integer len = object.getInteger("len");
		logger.info("len:"+len);
		//4、获取total len的长度，即有效载荷长度
		totalLen = object.getInteger("total");
		logger.info("total len:"+totalLen);
		//5、获取数据保存的位置
		int index = object.getInteger("index");
		if (recvData == null){
			recvData = new byte[totalLen*2];
		}
		//6、将message字段的数据保存到recvData中
		byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
		System.arraycopy(bytes, 0, recvData, index, message.length());
		recvLen += message.length();
		logger.info("recvlen:" + recvLen);
	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
		logger.info("发布成功...");
	}

	public byte[] sendAndRecv(ComService comService, int millsecond, boolean reSend){
		//制作发送数据帧
		byte[] sendBytes = new byte[60];
		if (!reSend) {
			frameCount++;
		}
		if (frameCount > MAX_FRAME_COUNT){
			frameCount = 1;
		}
		int frameSize = Frame.madeSendFrame(comService, sendBytes, reSend, frameCount);
		String message = ByteUtil.encode(sendBytes, frameSize);
		String pubTopic = mqtt.getProductKey() + "/" + mqtt.getDeviceName() + "/user/update";
		JSONObject object= new JSONObject();
		object.put("message", message);
		String str1 = object.toJSONString();
		//发送数据
		publish(pubTopic, str1, 1,false);
		//等待数据接收完成
		int timer = 0;
		int sleepTime = 500;
		while(true){
			if (recvLen > 0 && recvLen/2 == totalLen){
				break;
			}
			try {
				Thread.sleep(sleepTime);
				timer+=sleepTime;
				//超时时间
				if (timer == millsecond){
					logger.info("time out...");
					recvData = null;
					recvLen = 0;
					return null;
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		//接收数据完成
		ByteUtil.HexStr2Hex(recvData, recvLen);
		int payloadLen = ((recvData[2] & 0xF) << 4 | (recvData[3] & 0xF))
						| (((recvData[4] & 0xF) << 4 | (recvData[5] & 0xF)) << 8);
		System.out.println("payloadLen : "+payloadLen);
		payloadLen += 6;
		//接收的数据错误
		if (payloadLen > MAX_FRAME_LENGTH){
			recvData = null;
			recvLen = 0;
			return null;
		}
		payload = new byte[payloadLen];
		ByteUtil.HexBytes2Bytes(payload, 0, recvData, recvLen);
		//清空接收的数据
		recvData = null;
		recvLen = 0;
		//解析有效载荷，CRC
		if (Frame.isCurrentFrame(payload, frameCount) && Frame.isValidFrame(payload, payloadLen)){
			return payload;
		}
		return null;
	}
}
